/**
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.nhncorp.neptune.changelog.pipe;

import java.io.BufferedOutputStream;
import java.io.DataOutputStream;
import java.io.File;
import java.io.FileNotFoundException;
import java.io.FileOutputStream;
import java.io.FilenameFilter;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.channels.FileChannel;
import java.util.HashMap;
import java.util.concurrent.atomic.AtomicInteger;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

import com.nhncorp.neptune.common.TimeLog;

/**
 * @author sangchul
 * 하나의 ChangeLogFileChannel 인스턴스는 Tablet하나를 가리킨다.
 * 따라서 만일 하나의 tablet에 동시에 변경 요청이 수신 될 경우 
 * 두개 이상의 Pipe가 동시에 하나의 ChangeLogFileChannel 인스턴스에 접근할 수 있다. 
 * 따라서 적절한 동기화가 필요하다. 이를테면 쓰기 작업 시 seq 번호를 부여해서 
 * 쓰기 순서도 맞춰야 한다. 
 */
public class ChangeLogFileChannel {
  private static final Log LOG = LogFactory.getLog(ChangeLogFileChannel.class);

  private static HashMap<String, ChangeLogFileChannel> channelMap = new HashMap<String, ChangeLogFileChannel>();

  public static synchronized ChangeLogFileChannel createChannel(String tabletName, String path,
      String fileName) throws IOException {
    LOG.debug("channelMap, key : " + tabletName + fileName);
    ChangeLogFileChannel ch = channelMap.get(tabletName + fileName);

    if (ch == null) {
      ch = new ChangeLogFileChannel(tabletName, path, fileName);
	  LOG.info("create file channel. tabletName[" + tabletName + "], fileName[" + fileName + "]");
      channelMap.put(tabletName + fileName, ch);
    }

	ch.numPipes.incrementAndGet();
    return ch;
  }

  public static synchronized void closeChannel(ChangeLogFileChannel ch) throws IOException {
    int refCount = ch.numPipes.decrementAndGet();
    if (refCount <= 0) {
      String key = ch.tabletName + ch.logFileName;
      try {
	    LOG.info("close file channel. key[" + key + "]");
        ch.internalClose();
      } finally {
        if (channelMap.remove(key) == null) {
          LOG.warn("tried to remove not existing channel with name [" + key + "]");	
	    }
      }
      
      if (refCount != 0) {
        LOG.warn("wrong refCount [" + refCount + "] key [" + key + "]");
      }
    }
  }

  public static synchronized void backup(String tabletName, String path, String fileName) throws IOException {
    ChangeLogFileChannel ch = createChannel(tabletName, path, fileName);
    try {
	  ch.backup();
	} finally {
      closeChannel(ch);
    }
  }

  DataOutputStream dos;
  FileChannel channel;

  long rollbackIndex;
  final String logFileFullPath;
  final String logFileName;
  final String tabletName;
  final ByteBuffer lengthBuffer = ByteBuffer.allocate(4);
  final File parentDir;
  boolean fileConfirmed = false;
  AtomicInteger numPipes = new AtomicInteger(0);

  private ChangeLogFileChannel(String tabletName, String path, String fileName) throws IOException {
    this.tabletName = tabletName;
    parentDir = new File(path + tabletName);

    if (parentDir.exists() == false && parentDir.mkdirs() == false) {
      throw new IOException("Making directory to " + parentDir.getAbsolutePath() + " is fail");
    }

    logFileName = fileName;
    logFileFullPath = parentDir.getAbsolutePath() + File.separator + logFileName;

    init();
  }

  synchronized void init() throws IOException {
    FileOutputStream fos = null; 

    try {
      fos = new FileOutputStream(logFileFullPath, true);
    } catch(FileNotFoundException e) {
      if (parentDir.exists() == false && parentDir.mkdirs() == false) {
        throw new IOException("Making directory to " + parentDir.getAbsolutePath() + " is fail");
      }
      fos = new FileOutputStream(logFileFullPath, true);
    }
    
    dos = new DataOutputStream(new BufferedOutputStream(fos, 8192));
    channel = fos.getChannel();
	//channel.force(false);
    rollbackIndex = channel.size();
    channel.position(rollbackIndex);
  }

  public synchronized boolean isOpen() {
    if (channel == null) {
      return false;
    }

    return channel.isOpen();
  }

  ByteBuffer headerBuf = ByteBuffer.allocate(4);

  public synchronized void write(ByteBuffer[] bufferArray) throws IOException {
    if (channel == null) {
      LOG.warn("attempting to write log to already closed channel [" + logFileFullPath + "], but reinit");
      init();
      numPipes.set(1);
    }

    fileConfirmed = false;
    int size = 0;
    try {
      // write total length of all the buffers
      //dos.writeInt(getTotalLengthOf(bufferArray));
      
      headerBuf.putInt(getTotalLengthOf(bufferArray));
	  headerBuf.flip();
      channel.write(headerBuf);
	  headerBuf.clear();
      size += 4;

	  /*
      for (ByteBuffer buf : bufferArray) {
        byte[] array = new byte[buf.limit()];
        size += array.length;
        buf.get(array);
        dos.write(array);
      }
	  */

      long before = System.currentTimeMillis();
      for (ByteBuffer buf : bufferArray) {
        size += buf.limit();
        channel.write(buf);
      }
      long after = System.currentTimeMillis();

	  /*
      long before = System.currentTimeMillis();
      dos.flush();
      long after = System.currentTimeMillis();
	  */
      TimeLog.checkTime(before, 100, "channel.write(buf)");
      
//      if (after - before >= 1000) {
//        LOG.info("too long to flush log file. size : " + size + " bytes, time :  " + (after - before) + "ms, tablet[" + tabletName + "]");
//      }
    } catch(IOException e) {
      LOG.warn("fail to write log. full path [" + logFileFullPath + "]", e);
      internalClose();
      throw e;
    }
  }
  
  /*
  public void write(byte[] byteArray) throws IOException {
    if (channel == null) {
      LOG.warn("attempting to write log to already closed channel [" + logFileFullPath + "], but reinit");
      init();
    }

    fileConfirmed = false;
    int size = 0;
    try {
      // write total length of all the buffers
      dos.writeInt(byteArray.length);
      size += 4;

      dos.write(byteArray);
      size += byteArray.length;
      
      long before = System.currentTimeMillis();
      dos.flush();
      long after = System.currentTimeMillis();
      
      if (after - before >= 1000) {
        LOG.warn("It takes too long time to flush log file. size : " + size + " bytes, time :  " + (after - before) + "ms");
      }
    } catch(IOException e) {
      LOG.warn("fail to write log. full path [" + logFileFullPath + "]", e);
      internalClose();
      throw e;
    }
  }
  */

  private int getTotalLengthOf(ByteBuffer[] bufferArray) {
    int totalLength = 0;

    for (ByteBuffer buf : bufferArray) {
      totalLength += buf.limit();
    }

    return totalLength;
  }

  /*
  public synchronized void rollback() throws IOException {
    if (channel == null) {
      LOG.info("file channel to [" + logFileFullPath + "]is already closed. cannot rollback");
      return;
    }

    if (rollbackIndex < 0) {
      LOG.debug("nothing be rollbacked");
      return;
    } else if (rollbackIndex == new File(logFileFullPath).length()) {
      LOG.debug("TX has not yet been applied. there is no content being rollbacked");
      return;
    }

    if (channel != null && channel.isOpen()) {
      channel.truncate(rollbackIndex);
      channel.force(false);

      FileOutputStream fos = new FileOutputStream(logFileFullPath, true);
      dos = new DataOutputStream(fos);

      LOG.info(logFileFullPath + " is truncated to pos : " + rollbackIndex);
    } else {
      RandomAccessFile file = new RandomAccessFile(logFileFullPath, "rwd");
      FileChannel ch = file.getChannel();
      ch.truncate(rollbackIndex);
      ch.close();
      file.close();
      LOG.info(logFileFullPath + " is truncated to pos: " + rollbackIndex);
    }
  }

  public synchronized void resetRollbackIndex() {
    if (isOpen() && fileConfirmed == false) {
      fileConfirmed = true;

      try {
        long curSize = channel.size();

        if (curSize != rollbackIndex) {
          LOG.debug("file size : " + curSize + " is confirmed");
          rollbackIndex = curSize;
        }
      } catch (IOException e) {
        LOG.debug("cannot reset rollback index due to " + e);
      }
    }
  }
  */
  
  /*
  public synchronized void close() throws IOException {
    try {
      if (--numPipes <= 0) {
        internalClose();
        removeChannel(tabletName, logFileName);
        LOG.debug("close change log file channel path [" + logFileFullPath + "]");
      } else {
        LOG.debug("skip closing change log file [" + logFileFullPath + "] since it is shared with other pipes");
      }
    } catch (IOException e) {
      LOG.warn("closing file is fail", e);
      throw e;
    }
  }
  */

  private void internalClose() throws IOException {
    if (isOpen()) {
      channel.close();
      dos.close();
      channel = null;
      dos = null;
    }
  }

  public synchronized void removeFile() {
    try {
      internalClose();
    } catch (IOException e) {
      LOG.warn("closing file before removing is fail", e);
    }

    File file = new File(logFileFullPath);

    if (file.exists()) {
      if (file.delete()) {
        LOG.info("change log file [" + file.getPath() + "] is deleted");
      } else {
        LOG.warn("tried to delete change log file but it is NOT deleted.");
      }
    } else {
      LOG.info("attemped to remove file [" + file.getPath() + "], but not exists");
    }
  }

  public synchronized long getFileSize() throws IOException {
    File file = new File(logFileFullPath);
    return file.length();
  }

  public synchronized void backup() throws IOException {
    long beforeClose = System.currentTimeMillis();
    try {
      internalClose();
    } catch (IOException e) {
      LOG.warn("closing file before backing up is fail", e);
    }

    long beforeRename = System.currentTimeMillis();
    File file = new File(logFileFullPath);
    if (!file.renameTo(new File(parentDir, file.getName() + "_" + System.currentTimeMillis()))) {
      throw new IOException("fail to backup [" + file.getAbsolutePath() + "]");
	}

    long beforeInit = System.currentTimeMillis();
    init();

	long finish = System.currentTimeMillis();

	if ((finish - beforeInit) >= 1000) {
		LOG.info("too long time to init : " + (finish - beforeInit) + "ms");
	} else if ((beforeInit - beforeRename) >= 1000) {
		LOG.info("too long time to rename : " + (beforeInit - beforeRename) + "ms");
	} else if ((beforeRename - beforeClose) >= 1000) {
		LOG.info("too long time to close : " + (beforeRename - beforeClose) + "ms");
	}
  }

  public synchronized void restoreLastLog() throws IOException {
    try {
      internalClose();
    } catch (IOException e) {
      LOG.warn("closing file in restoring file is fail", e);
    }

    File lastLogFile = null;

    long lastModified = 0l;
    for (File file : parentDir.listFiles()) {
      if (lastLogFile == null) {
        lastLogFile = file;
        continue;
      }

      if (lastModified < file.lastModified() && lastLogFile.getName().equals(logFileName) == false) {
        lastModified = file.lastModified();
        lastLogFile = file;
      }
    }

    lastLogFile.renameTo(new File(parentDir, logFileName));

    init();
  }

  public synchronized void removeBackupFiles() {
    File[] fileList = parentDir.listFiles(new FilenameFilter() {
      public boolean accept(File dir, String name) {
        return name.equals(logFileName) == false;
      }
    });

    for (File file : fileList) {
      LOG.info("remove backup file [" + file.getAbsolutePath() + "]");
      if (!file.delete()) {
		LOG.warn("fail to remove backup file [" + file.getAbsolutePath() + "]");
	  }
    }
  }
}
